package com.neusoft.hifly.netty.socket.handler;

import java.net.InetSocketAddress;
import java.time.LocalDateTime;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.lang3.StringUtils;
import org.springframework.kafka.core.KafkaTemplate;

import com.neusoft.hifly.core.log.Log;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class SocketChatHandler extends SimpleChannelInboundHandler<String> {
	// IP与信道的对应关系
	private ThreadLocal<ConcurrentHashMap<String, Channel>> channelMap = new ThreadLocal<ConcurrentHashMap<String, Channel>>();

	// kafka
	private KafkaTemplate<String, Object> kafkaTemplate;

	public SocketChatHandler(KafkaTemplate<String, Object> kafkaTemplate) {
		super();
		this.kafkaTemplate = kafkaTemplate;
	}

	/**
	 * 客户端发消息会触发
	 */
	@Override
	public void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {

		msg = StringUtils.trimToEmpty(msg);

		final String ip = getIP(ctx);

		Log.info("[" + ip + "]收到消息：" + msg);

		sendMsg(ip, "[服务器在]" + LocalDateTime.now() + "接受到消息（已发送给Kafka）, 消息为：" + msg);

		kafkaTemplate.send("micro-netty", msg);
	}

	/**
	 * 客户端连接会触发
	 */
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		final String ip = getIP(ctx);
		//添加channel信息
		putChannel(ip, ctx.channel());
		Log.info("[" + ip + "]已连接。。。");
		sendMsg(ip, "[connect sucess ...]" + LocalDateTime.now());
	}

	/**
	 * 客户端断开连接会触发
	 */
	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		final String ip = getIP(ctx);
		//删除失效的channel
		removeChannel(ip);
		ctx.close();
		Log.info("[" + ip + "]已断开。。。");
	}

	/**
	 * 发生异常触发
	 */
	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable t) throws Exception {
		final String ip = getIP(ctx);
		Log.info("[" + ip + "]发生异常：" + t);
		t.printStackTrace();
		ctx.close();
	}

	/**
	 * 获取客户端IP地址
	 */
	private String getIP(final ChannelHandlerContext ctx) {
		final InetSocketAddress insocket = (InetSocketAddress) ctx.channel().remoteAddress();
		final String clientIP = insocket.getAddress().getHostAddress();
		return clientIP;
		//		final String socketString = ctx.channel().remoteAddress().toString();
		//		final int index = socketString.indexOf(":");
		//		return socketString.substring(1, index);
	}

	// 添加信道
	private void putChannel(String ip, Channel channel) {
		getChannel().put(ip, channel);
	}

	// 删除信道
	private void removeChannel(String ip) {
		getChannel().remove(ip);
	}

	// 发送消息
	private void sendMsg(String ip, String msg) {
		Channel channel = getChannel().get(ip);
		if (channel != null) {
			channel.writeAndFlush(msg);
		}
	}

	// 获取客户端列表
	private ConcurrentHashMap<String, Channel> getChannel() {
		ConcurrentHashMap<String, Channel> map = this.channelMap.get();
		if (map == null) {
			map = new ConcurrentHashMap<>();
			channelMap.set(map);
		}
		return map;
	}
}